/*
 * @Descripttion: MQTT客户端
 * @Author: wuguangbin
 * @email:  guangbin.wu@hollyland-tech.com
 * @Date: 2023-11-27 17:32:49
 */

#include "MqttClient.hxx"
#include "Log.hxx"
#include "ProductDef.hxx"

#define SEND_QUEUE_MAX_SIZE 256 // 发送队列最大支持数量
#define RECONECT_TIME 3000      // 断连重连时间
#define RESTART_TIME 5000       // 断连重启客户端时间

/**
 * @brief: 状态机初始化:空闲 安全认证 连接 重连 传输 断连
 * @param {*}
 * @return {*}
 */
const CMqttClient::MapStateMachineFunc_t CMqttClient::m_MapStateMachineFunc = {
    ValueType("Idle", &CMqttClient::_IdleState),
    ValueType("Auth", &CMqttClient::_AuthState),
    ValueType("Connect", &CMqttClient::_ConnectState),
    ValueType("ReConnect", &CMqttClient::_ReconnectState),
    ValueType("DisConnect", &CMqttClient::_DisconnectState),
    ValueType("Transfer", &CMqttClient::_TransferState),
};

CMqttClient::CMqttClient()
    : mosquittopp(NULL, true), CThread(false, 0, 0, "CMqttClient"), m_qos(1), m_Mutex("CMqttClient"), m_StateMachine("Idle"), m_strHost("127.0.0.1"), m_iPort(1883), m_StateChangeFun(nullptr), m_RecvMessageFun(nullptr), m_StateMachineMutex("StateMachine"), m_bStart(false), m_bLog(true), m_MutexMqttSubmit("CMqttClientMqttSubmit")
{
    m_bConnect.store(false);
    m_bSendFlag.store(false);
}

CMqttClient::~CMqttClient()
{
}

bool CMqttClient::Init(const char *host, int port, int qos)
{
    // 1、初始化broker信息
    m_strHost.assign(host);
    m_iPort = port;
    m_qos = qos;
    // 2、初始化Mqtt库
    m_vecSubTopic.reserve(1024);
    m_VecRetainMsg.reserve(SEND_QUEUE_MAX_SIZE);
    mosqpp::lib_init();
    int major, minor, revision;
    mosqpp::lib_version(&major, &minor, &revision);
    DBG_MQTT("init mosqpp ver: %d.%d.%d", major, minor, revision);
    return CreateThread();
}

bool CMqttClient::CleanUp()
{
    DestroyThread(true);
    mosqpp::lib_cleanup();
    DBG_MQTT("cleanup mosqpp lib");
    m_VecRetainMsg.clear();
    m_vecSubTopic.clear();
    return true;
}

bool CMqttClient::CheckBrokerAlive(const char *host, int port, int qos)
{
    bool bRet = false;
    int iRet = MOSQ_ERR_SUCCESS;
    iRet = username_pw_set("hollyland", "123456");
    if (MOSQ_ERR_SUCCESS == iRet)
    {
        iRet = connect(host, port, 5);
        if (MOSQ_ERR_SUCCESS == iRet)
        {
            DBG_MQTT("Check broke %s: %d is Alive", host, port);
            bRet = true;
        }
        else
        {
            DBG_MQTT("Check broke %s : %d is not Alive : [%d] %s", host, port, iRet, mosqpp::strerror(iRet));
            return false;
        }
    }
    else
    {
        ERR_MQTT("auth broke %s: %d fail : %s, try against", host, port, mosqpp::strerror(iRet));
        return false;
    }

    if (bRet)
    {
        iRet = disconnect();
        if (MOSQ_ERR_SUCCESS == iRet)
        {
            // DBG_MQTT("disconnect broke %s: %d ok", host, port);
        }
        else
        {
            ERR_MQTT("Check broke %s : %d fail : [%d] %s", host, port, iRet, mosqpp::strerror(iRet));
        }
    }
    return bRet;
}

bool CMqttClient::ReInit(const char *host, int port, int qos)
{
    m_qos = qos;
    if (m_strHost.compare(string(host)) == 0 && m_iPort == port)
    {
        DBG_MQTT("host /port is same");
        return false;
    }
    else
    {
        m_strHost.clear();
        m_strHost.assign(host);
        m_iPort = port;
        DBG_MQTT("ReInit connect broker  %s : %d", m_strHost.c_str(), m_iPort);
    }
    return true;
}

/**
 * @brief: 状态机启动运行
 * @param {void}
 * @return {true: 成功 false: 失败}
 */
bool CMqttClient::Start()
{
    if (!m_bStart)
    {
        DBG_MQTT("start MqttClient");
        CGuard guard(m_StateMachineMutex);
        m_queueNextStateMachine.emplace(string("Auth"));
        m_bStart = true;
    }
    else
    {
        ERR_MQTT("CMqttClient is already start ");
        return false;
    }
    return true;
}

/**
 * @brief: 状态机停止运行
 * @param {void}
 * @return {true: 成功 false: 失败}
 */
bool CMqttClient::Stop()
{
    if (m_bStart)
    {
        DBG_MQTT("stop MqttClient");
        CGuard guard(m_StateMachineMutex);
        m_bSendFlag.store(false);
        for (int i = 0; i < 10; i++)
        {
            m_queueNextStateMachine.emplace(string("Idle"));
        }
        m_queueNextStateMachine.emplace(string("DisConnect"));
        m_bStart = false;
    }
    else
    {
        ERR_MQTT("CMqttClient is already stop ");
        return false;
    }
    return true;
}

/**
 * @brief: 状态机重新运行
 * @param {void}
 * @return {true: 成功 false: 失败}
 */
bool CMqttClient::Restart()
{
    DBG_MQTT("restart MqttClient");
    XSleep(RESTART_TIME);
    CGuard guard(m_StateMachineMutex);
    m_queueNextStateMachine.emplace(string("DisConnect"));
    m_queueNextStateMachine.emplace(string("Auth"));
    return true;
}

/**
 * @brief: SDK的will遗言设置，MQTT客户端下线后，broke发布遗嘱通知其他client，必须要在Start前调用
 * @param {strTopic : 遗嘱topic  strMsg: 消息体}
 * @return {0:成功  其他:失败}
 */
int CMqttClient::WillSet(const string &strTopic, const string &strMsg)
{
    if (m_bConnect.load())
    {
        ERR_MQTT("will set must before connect ");
        return -1;
    }
    // DBG_MQTT("WillSet topic :%s msg :%s", strTopic.c_str(), strMsg.c_str());
    return will_set(strTopic.c_str(), strMsg.size(), strMsg.c_str(), m_qos, false);
}

/**
 * @brief: 清空will
 * @param {null}
 * @return {0:成功  其他:失败}
 */
int CMqttClient::WillClear()
{
    if (m_bConnect.load())
    {
        ERR_MQTT("will set must before conect ");
        return -1;
    }
    return will_clear();
}

/**
 * @brief: 发布消息, 连接上broker再调用发布
 * @param {strTopic : 发布topic  strMsg: 消息体 retain: retain特性使能}
 * @return {0:成功  其他:失败}
 */
int CMqttClient::Publish(const string &strTopic, const string &strMsg, bool retain)
{
    if (strTopic.empty() || strMsg.empty())
    {
        ERR_MQTT("strTopic / strMsg is null");
        return -1;
    }

    CGuard guard(m_Mutex);
    // Retain保留数据缓存到Retain消息队列
    if (retain)
    {
        if (m_VecRetainMsg.size() < SEND_QUEUE_MAX_SIZE)
        {
            // DBG_MQTT("retain Publish msg in queue ok:  %d", m_VecSendMsg.size());
        }
        else
        {
            m_VecRetainMsg.erase(m_VecRetainMsg.begin());
        }

        SendMessage msg;
        msg.iMid = m_stMid.iPublishMid;
        msg.strTopic = strTopic;
        msg.strMsg = strMsg;
        msg.bRetain = retain;
        if (std::find(m_VecRetainMsg.begin(), m_VecRetainMsg.end(), msg) == m_VecRetainMsg.end())
        {
            m_VecRetainMsg.emplace_back(msg);
            INF("wgb retain msg ：[%s] add queue ok", msg.strTopic.c_str());
        }
        else
        {
            DBG_MQTT("wgb retain msg ：[%s] is exsit ,no need add", msg.strTopic.c_str());
        }
    }

    if (!m_bConnect.load() || !m_bSendFlag.load())
    {
        // broker离线状态下消息不处理
        DBG_MQTT("wgb offline topic not process topic: %s  messageSize: [%ld], m_bConnect = %d, m_bSendFlag = %d",
                 strTopic.c_str(), strMsg.size(), m_bConnect.load(), m_bSendFlag.load());
        return -1;
    }
    else
    {
        // broker在线状态下直接发布
        int iRet = publish(&m_stMid.iPublishMid, strTopic.c_str(), strMsg.size(), strMsg.c_str(), m_qos, retain);
        DBG_MQTT("mid: [%d] send topic : %s send messageSize: [%ld] ", m_stMid.iPublishMid, strTopic.c_str(),
                 strMsg.size());
        return iRet;
    }
}

/**
 * @brief: 订阅消息
 * @param {strTopic : 订阅topic }
 * @return {0:成功  其他:失败}
 */
int CMqttClient::Subscribe(const string &strTopic)
{
    if (strTopic.empty())
    {
        ERR_MQTT("strTopic is null");
        return -1;
    }

    int iRet = MOSQ_ERR_SUCCESS;
    CGuard guard(m_Mutex);
    if (!m_vecSubTopic.empty())
    {
        for (const auto &iter : m_vecSubTopic)
        {
            if (iter.compare(strTopic) == 0)
            {
                DBG_MQTT("topic : %s is arleady subscribe", strTopic.c_str());
                return -1;
            }
        }
    }

    // 离线状态下先缓存订阅请求
    m_vecSubTopic.emplace_back(strTopic);

    // 在线状态下立即发送订阅
    if (m_bConnect.load() && m_bSendFlag.load())
    {
        iRet = subscribe(&m_stMid.iSubMid, strTopic.c_str(), m_qos);
        DBG_MQTT("mid: [%d] subscribe topic : %s ", m_stMid.iSubMid, strTopic.c_str());
    }
    return iRet;
}

/**
 * @brief: 解除订阅消息
 * @param {strTopic : 订阅topic }
 * @return {0:成功  其他:失败}
 */
int CMqttClient::Unsubscribe(const string &strTopic)
{
    if (strTopic.empty())
    {
        ERR_MQTT("strTopic is null");
        return -1;
    }

    int iRet = MOSQ_ERR_SUCCESS;
    bool bUnsubscrib = false;
    CGuard guard(m_Mutex);
    if (m_vecSubTopic.empty())
    {
        DBG_MQTT("topic vec is empty");
        return -1;
    }

    // 删除订阅列表
    for (auto iter = m_vecSubTopic.begin(); iter != m_vecSubTopic.end();)
    {
        if ((*iter).compare(strTopic) == 0)
        {
            iter = m_vecSubTopic.erase(iter);
            bUnsubscrib = true;
        }
        else
        {
            iter++;
        }
    }

    // 在线状态下且已经订阅了的topic才触发解除订阅
    if (m_bConnect.load() && bUnsubscrib && m_bSendFlag.load())
    {
        iRet = unsubscribe(&m_stMid.iUnsubMid, strTopic.c_str());
        DBG_MQTT("mid: [%d] unsubscribe topic : %s ", m_stMid.iUnsubMid, strTopic.c_str());
    }
    return iRet;
}

/**
 * @brief: MQTT客户端连接状态回调
 * @param {fun : StateChangeCallback函数指针，不允许阻塞和耗时操作}
 * @return {void }
 */
void CMqttClient::RegisterStateCallback(const StateChangeCallback &fun)
{
    m_StateChangeFun = fun;
}

void CMqttClient::UnRegisterStateCallback()
{
    m_StateChangeFun = nullptr;
}

/**
 * @brief: MQTT客户端主题数据接收回调，注意fun不可以阻塞
 * @param {fun : StateChangeCallback函数指针}
 * @return {void }
 */
void CMqttClient::RegisterRecvMessageCallback(const RecvMessageCallback &fun)
{
    m_RecvMessageFun = fun;
}

void CMqttClient::UnRegistRecvMessageCallback()
{
    m_RecvMessageFun = nullptr;
}

string CMqttClient::_IdleState()
{
    XSleep(100);
    return string("Idle");
}

string CMqttClient::_AuthState()
{
    DBG("oyls _AuthState........");
    string strRet;
    int iRet = MOSQ_ERR_SUCCESS;
    iRet = username_pw_set("hollyland", "123456");
    if (MOSQ_ERR_SUCCESS == iRet)
    {
        DBG("oyls Configure Username and Password Successfull [IP %s: %d] ", m_strHost.c_str(), m_iPort);
        strRet = string("Connect");
    }
    else
    {
        ERR("oyls auth broke %s: %d fail : %s, try against", m_strHost.c_str(), m_iPort, mosqpp::strerror(iRet));
        XSleep(RECONECT_TIME);
        strRet = string("Auth");
    }
    DBG("oyls _AuthState end........");
    return strRet;
}

string CMqttClient::_ConnectState()
{
    DBG("oyls _ConnectState........");
    string strRet;
    int iRet = MOSQ_ERR_SUCCESS;
    DBG("oyls _ConnectState host : %s port %d", m_strHost.c_str(), m_iPort);
    iRet = connect_async(m_strHost.c_str(), m_iPort, 60);
    if (MOSQ_ERR_SUCCESS == iRet)
    {
        DBG("oyls connect broke %s: %d ok", m_strHost.c_str(), m_iPort);
        strRet = string("Transfer");
    }
    else
    {
        ERR("oyls connect broke %s : %d fail : [%d] %s", m_strHost.c_str(), m_iPort, iRet, mosqpp::strerror(iRet));
        strRet = string("ReConnect");
    }
    DBG("oyls _ConnectState end........");
    return strRet;
}

string CMqttClient::_TransferState()
{
    DBG("oyls _TransferState........");
    string strRet;
    int iRet = MOSQ_ERR_SUCCESS;
    iRet = loop_start();
    if (MOSQ_ERR_SUCCESS != iRet)
    {
        ERR("oyls TransferState fail : [%d] %s", iRet, mosqpp::strerror(iRet));
        Restart();
    }
    DBG("oyls _TransferState end........");
    return string("Idle");
}

string CMqttClient::_ReconnectState()
{
    DBG("oyls _ReconnectState........");
    string strRet;
    strRet = string("Auth");
    XSleep(RECONECT_TIME);
    DBG("oyls _ReconnectState end........");
    return strRet;
}

string CMqttClient::_DisconnectState()
{
    DBG("oyls _DisconnectState........");
    string strRet;
    int iRet = 0;

    if (m_bConnect.load())
    {
        // 1. 先发送断开连接请求
        iRet = disconnect();
        if (MOSQ_ERR_SUCCESS == iRet)
        {
            DBG("oyls disconnect request sent to broker %s : %d", m_strHost.c_str(), m_iPort);
            // 2. 给一个短暂的时间让断开连接消息发送完成
            XSleep(500);
        }
        else
        {
            ERR("oyls disconnect request fail: [%d]: %s", iRet, mosqpp::strerror(iRet));
        }

        // 3. 更新连接状态
        m_bConnect.store(false);
        m_bSendFlag.store(false);

        // 4. 停止消息循环，使用超时保护
        iRet = loop_stop(false); // 先尝试优雅停止
        if (MOSQ_ERR_SUCCESS != iRet)
        {
            ERR("oyls graceful loop_stop fail: [%d]: %s", iRet, mosqpp::strerror(iRet));
            // 如果优雅停止失败，等待一段时间后强制停止
            XSleep(1000);
            iRet = loop_stop(true);
            if (MOSQ_ERR_SUCCESS != iRet)
            {
                ERR("oyls force loop_stop fail: [%d]: %s", iRet, mosqpp::strerror(iRet));
            }
        }
        else
        {
            DBG("oyls loop stopped successfully for broker %s : %d", m_strHost.c_str(), m_iPort);
        }
    }

    strRet = string("Idle");
    DBG("oyls _DisconnectState end........");
    return strRet;
}

bool CMqttClient::_EnterState(const string &strSta)
{
    MapStateMachineFunc_t::const_iterator it = m_MapStateMachineFunc.find(strSta);
    if (it == m_MapStateMachineFunc.end())
    {
        ERR("oyls not surpport StateMachine : [%s] ", strSta.c_str());
        return false;
    }
    if (m_StateMachine.compare(strSta))
    {
        DBG("oyls CMqttClient StateMachine change from : %s ----> %s", m_StateMachine.c_str(), strSta.c_str());
        m_StateMachine = strSta;
    }
    return true;
}
/**
 * @brief 检测是否有下一状态机切换任务存在
 *
 * @return 不空，返回下一个要执行的状态
 * @return 空，没有下一个状态
 */
string CMqttClient::_GetNextState()
{
    string strNextStateMachine;
    CGuard guard(m_StateMachineMutex);
    if (!m_queueNextStateMachine.empty())
    {
        strNextStateMachine = m_queueNextStateMachine.front();
        DBG("exec NextStateMachine: %s", strNextStateMachine.c_str());
        m_queueNextStateMachine.pop();
    }
    else
    {
        // DBG_MQTT("no hase NextState");
        strNextStateMachine.clear();
    }
    return strNextStateMachine;
}

void CMqttClient::on_connect(int rc)
{
    m_bConnect.store(true);
    m_bSendFlag.store(true);
    DBG("oyls on_connect: rc = [%d] : %s", rc, mosqpp::strerror(rc));

    // 连接成功通知上层
    ClientState state;
    state.bOnline = true;
    if (m_StateChangeFun)
    {
        m_StateChangeFun(&state, sizeof(state));
    }

    // 连接broker成功
    if (rc == 0)
    {
        // 发送订阅列表
        CGuard guard(m_Mutex);
        if (!m_vecSubTopic.empty())
        {
            DBG_MQTT("sync subscribe to broke %s: %d", m_strHost.c_str(), m_iPort);
            for (const auto &iter : m_vecSubTopic)
            {
                subscribe(&m_stMid.iSubMid, iter.c_str(), m_qos);
            }
        }

        // Retain保留数据发送
        if (!m_VecRetainMsg.empty())
        {
            DBG_MQTT("publish Retain message to broke %s: %d", m_strHost.c_str(), m_iPort);
            for (auto &iter : m_VecRetainMsg)
            {
                publish(&m_stMid.iPublishMid, iter.strTopic.c_str(), iter.strMsg.size(), iter.strMsg.c_str(), m_qos,
                        iter.bRetain);
                INF("wgb Retain mid: [%d] send topic: [%s] messageSize: %ld ", m_stMid.iPublishMid,
                    iter.strTopic.c_str(), iter.strMsg.size());
            }
        }
    }
}

void CMqttClient::on_disconnect(int rc)
{
    m_bConnect.store(false);
    DBG("oyls on_disconnect: rc = [%d] : %s", rc, mosqpp::strerror(rc));
    // 离线通知上层
    ClientState state;
    state.bOnline = false;
    if (m_StateChangeFun)
    {
        m_StateChangeFun(&state, sizeof(state));
    }
}

void CMqttClient::on_publish(int mid)
{
    DBG_MQTT("on_publish mid: [%d] ", mid);
}

void CMqttClient::on_subscribe(int mid, int qos_count, const int *granted_qos)
{
    if (granted_qos)
    {
        DBG_MQTT("on_subscribe mid: [%d] qos_count: %d granted_qos: %d", mid, qos_count, *granted_qos);
    }
    else
    {
        ERR_MQTT("granted_qos is null");
    }
}

void CMqttClient::on_message(const struct mosquitto_message *message)
{
    if (message && m_bSendFlag.load())
    {
        DBG_MQTT("mid: [%d] recv topic: %s messageSize: [%d], qos : %d, retain : %d", message->mid, message->topic,
                 message->payloadlen, message->qos, message->retain);

        RecvMessage msg;
        msg.iMid = message->mid;
        msg.strTopic.assign(message->topic);
        msg.strMsg.assign((char *)message->payload);

        if (m_RecvMessageFun)
        {
            m_RecvMessageFun(&msg, sizeof(msg));
        }
    }
    else
    {
        ERR_MQTT("message is null, m_bSendFlag = %d", m_bSendFlag.load());
    }
}

void CMqttClient::on_log(int level, const char *str)
{
    if (m_bLog)
    {
        if (str)
        {
            DBG_MQTT("[%d] on_log : %s", level, str);
        }
        else
        {
            ERR_MQTT("[%d] str is null", level);
        }
    }
    else
    {
        return;
    }
}

/**
 * @brief: broker返回的错误状态（未清楚）
 * @param {*}
 * @return {*}
 */
void CMqttClient::on_error()
{
    ERR_MQTT("broker on_error");
    CGuard guard(m_StateMachineMutex);
    m_queueNextStateMachine.emplace(string("ReConnect"));
}

void CMqttClient::on_unsubscribe(int mid)
{
    DBG_MQTT("on_unsubscribe mid: %d ", mid);
}

bool CMqttClient::DeleteRetainMsg(const string &strTopic)
{
    // 删除订阅列表
    //  INF("############wgb DeleteRetainMsg : %s ####################", strTopic.c_str());
    CGuard guard(m_Mutex);
    bool bRet = false;
    for (auto iter = m_VecRetainMsg.begin(); iter != m_VecRetainMsg.end();)
    {
        if (iter->strTopic.compare(strTopic) == 0)
        {
            iter = m_VecRetainMsg.erase(iter);
            INF("wgb DeleteRetainMsg : %s delete ok ", strTopic.c_str());
            bRet = true;
        }
        else
        {
            iter++;
        }
    }
    return bRet;
}

bool CMqttClient::EnableLog(bool bEnable)
{
    if (m_bLog != bEnable)
    {
        if (bEnable)
        {
            DBG_MQTT("start mqtt log ...");
        }
        else
        {
            DBG_MQTT("stop mqtt log ...");
        }
        m_bLog = bEnable;
    }
    return true;
}

void CMqttClient::ThreadProc()
{
    LOG_SYSTEM("Enter CMqttClient Thread, thread PID = %d, PPID = %d.", GetThreadID(), getppid());
    string strStateMachine;
    string strNextStateMachine;

    while (m_bLoop)
    {
        MapStateMachineFunc_t::const_iterator it = m_MapStateMachineFunc.find(m_StateMachine);
        if (it != m_MapStateMachineFunc.end())
        {
            strStateMachine = (this->*(it->second))();
            // 当前状态执行完成，切换下一状态的处理
            strNextStateMachine = _GetNextState();
            if (!strNextStateMachine.empty())
            {
                strStateMachine = strNextStateMachine;
            }
            _EnterState(strStateMachine);
        }
        else
        {
            ERR_MQTT("not surpport StateMachine %s", m_StateMachine.c_str());
        }
    }
}

void CMqttClient::PrintDebugInfo()
{
    DBG("m_strHost: %s, m_iPort:%d, m_qos:%d, bStart:%d, bLog:%d, bConnect:%d, bSendFlag%d, vecSubTopicSize:%ld, "
        "VecRetainMsgSize:%ld, StateMachine:%s",
        m_strHost.c_str(), m_iPort, m_qos, m_bStart, m_bLog, m_bConnect.load(), m_bSendFlag.load(),
        m_vecSubTopic.size(), m_VecRetainMsg.size(), m_StateMachine.c_str());

    // 打印队列中的所有元素
    CGuard guard(m_StateMachineMutex);
    DBG("Next State Machine Queue:");
    std::queue<string> tempQueue = m_queueNextStateMachine; // 创建副本以避免修改原队列
    while (!tempQueue.empty())
    {
        DBG("  %s", tempQueue.front().c_str());
        tempQueue.pop();
    }
}

int CMqttClient::GetMachineSize()
{
    CGuard guard(m_StateMachineMutex);
    // DBG("Next State Machine Queue Size: %ld", m_queueNextStateMachine.size());
    //  if ((m_queueNextStateMachine.size() == 1 && m_queueNextStateMachine.front().compare("Idle") == 0) {
    //      return 0;
    //  }
    return m_queueNextStateMachine.size();
}
